package com.lintf.bot.rabbitmq;

import com.alibaba.fastjson2.JSONObject;
import com.lintf.bot.domain.dto.BotGroupInviteListDTO;
import com.lintf.bot.rabbitmq.constant.Constants;
import com.lintf.bot.rabbitmq.delay.CommonMessageDelayService;
import com.lintf.bot.service.IBotGroupInviteService;
import com.lintf.bot.wechat.IWeChatNoticeService;
import com.lintf.bot.wechat.dto.NotificationsSendDTO;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author lintf
 */
@Slf4j
@Component
public class MessageConsumer extends CommonMessageDelayService {

    private static final int MAX_CONSUME_COUNT = 10;

    private Map<String, Integer> consumeCountMap = new HashMap<>();

    @Autowired
    private IWeChatNoticeService weChatService;

    @Autowired
    private IBotGroupInviteService botGroupInviteService;

    @RabbitListener(queues = Constants.LIKE_NOTIFICATION_QUEUE, ackMode = "MANUAL")
    public void receiveLikeNotification(String json, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Message message, Channel channel) throws IOException {
        try {
            NotificationsSendDTO dto = this.checkInit(json);
            if (dto == null) return;
            weChatService.sendLikeNotifications(dto);
            log.info("======队列消息处理完毕======");
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.info("消费异常：{}", e.getMessage());
            channel.basicNack(deliveryTag, false, false);
            sendDelayMessage(message, e);
        }
    }

    @RabbitListener(queues = Constants.NOTIFICATION_QUEUE, ackMode = "MANUAL")
    public void receiveNotification(String json, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Message message, Channel channel) throws IOException {
        try {
            NotificationsSendDTO dto = this.checkInit(json);
            if (dto == null) return;
            BotGroupInviteListDTO listDTO = new BotGroupInviteListDTO();
            listDTO.setGroupId(dto.getGroupId());
            listDTO.setIncludeOfOwn(false);
            listDTO.setExcludeOfUserId(dto.getCurrentUserId());
            dto.setSysUsers(botGroupInviteService.getGroupOfUser(listDTO));
            weChatService.sendNotifications(dto);
            log.info("======队列消息处理完毕======");
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.info("消费异常：{}", e.getMessage());
            channel.basicNack(deliveryTag, false, false);
            sendDelayMessage(message, e);
        }
    }


    @RabbitListener(queues = Constants.ITEM_NOTIFICATION_QUEUE, ackMode = "MANUAL")
    public void receiveItemNotificationQueue(String json, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Message message, Channel channel) throws IOException {
        try {
            NotificationsSendDTO dto = this.checkInit(json);
            if (dto == null) return;
            weChatService.sendItemNotifications(dto);
            log.info("======队列消息处理完毕======");
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.info("消费异常：{}", e.getMessage());
            channel.basicNack(deliveryTag, false, false);
            sendDelayMessage(message, e);
        }
    }

    /**
     * 初始化检测
     *
     * @param json 消息体
     * @return 实体
     */
    private NotificationsSendDTO checkInit(String json) {
        if (StringUtils.equals(json, "init")) {
            log.info("======初始化测试======");
            return null;
        }
        log.info("======消费者开始处理队列消息======");
        log.info("【正常队列】【" + System.currentTimeMillis() + "】收到正常队列消息：" + json);
        NotificationsSendDTO dto = JSONObject.parseObject(json, NotificationsSendDTO.class);
        dto.setAccessToken(weChatService.getAccessToken());
        return dto;
    }

    /**
     * 消息丢弃策略
     *
     * @param message 消息体
     * @param channel 信道
     * @param json    数据
     * @return
     */
    public boolean discardPolicy(Message message, Channel channel, String json) {
        String messageId = message.getMessageProperties().getMessageId();
        int consumeCount = consumeCountMap.getOrDefault(messageId, 0);

        if (consumeCount >= MAX_CONSUME_COUNT) {
            log.warn("消息已超过最大消费次数，将被丢弃：" + json);
            try {
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                log.warn("消息消费应答发生未知错误");
            }
            return true;
        } else {
            return false;
        }
    }
}
